package com.example.yckjbigdataservice.controller;

import com.alibaba.fastjson2.JSON;
import com.example.yckjbigdataservice.utils.HiveDataSourceUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@RequestMapping("/hive")
@RestController
@Slf4j
public class HiveController {

    /**
     * 定义Hive样例使用的SQL
     */
    private static final String[] SQLS = {
            "CREATE TABLE IF NOT EXISTS employees_infoa(id INT, age INT, name STRING)",
            "INSERT INTO employees_infoa VALUES(001, 31, 'SJK'),(002, 25, 'HS'),(003, 28, 'HT')",
            "SELECT * FROM employees_infoa",
            "DROP TABLE employees_infoa"
    };

    @Value("${custom.hive.config-path}")
    private String configPath;

    @GetMapping("/simple/query")
    public String hiveQuery(@RequestParam("where") String where) {
        // 查询数据
        //String querySql = "SELECT * FROM hive_std_16LZDR WHERE cq00027_DE00001 = cq00027_DE00002";
        int page = 1;
        int pageSize = 10;
        int start = (page - 1) * pageSize + 1;
        int end = page * pageSize;

        String querySql = "SELECT id, CQ00027_DE00001, CQ00027_DE00002 " +
                "FROM ( " +
                "  SELECT id, CQ00027_DE00001, CQ00027_DE00002, ROW_NUMBER() OVER (ORDER BY id) AS rn " +
                "  FROM hive_std_16LZDR WHERE CQ00027_DE00001 = ? " +
                ") t " +
                "WHERE rn BETWEEN " + start + " AND " + end;
        /*String querySql = "SELECT id, CQ00027_DE00001, CQ00027_DE00002 " +
                "  FROM hive_std_16LZDR WHERE CQ00027_DE00001 = ? ";*/

        //Class.forName("org.apache.hive.jdbc.HiveDriver");
        //String queryResult = HiveDataSourceUtil.executeQuery(connection, SQLS[2]);
        try (Connection conn = HiveDataSourceUtil.createConnection(configPath);
                 /*Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery(querySql)*/
             PreparedStatement pstmt = conn.prepareStatement(querySql) // 参数设置使用PreparedStatement
        ) {

            pstmt.setString(1, where);

            try (ResultSet rs = pstmt.executeQuery()) {
                List<Map<String, Object>> resultList = new ArrayList<>();
                ResultSetMetaData meta = rs.getMetaData();
                int colCount = meta.getColumnCount();

                while (rs.next()) {
                    Map<String, Object> row = new HashMap<>();

                    for (int i = 1; i <= colCount; i++) {
                        row.put(meta.getColumnName(i), rs.getObject(i));
                    }
                    log.info("每一行数据 = {}", JSON.toJSONString(row));

                    resultList.add(row);
                }
                // 打印结果
                resultList.forEach(item -> log.info("{}", JSON.toJSONString(item)));
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        return "hive 查询完毕success";
    }

    /**
     * 指定数据保存到不同格式
     */
    public void saveFile(){
        String jdbcUrl = "jdbc:hive2://hive-server:10000/default";
        String user = "hive";
        String password = "";

        try (Connection conn = DriverManager.getConnection(jdbcUrl, user, password);
             Statement stmt = conn.createStatement()) {

            // 通用压缩设置
            stmt.execute("SET hive.exec.compress.output=true");

            // ✳ 设置不同存储格式的压缩参数（根据你想写的格式选择其一）
            // ① Parquet 使用 GZIP（也支持 SNAPPY）
            stmt.execute("SET parquet.compression=GZIP");//SET parquet.compression=SNAPPY;

            // ② ORC 使用 ZLIB（也支持 NONE、SNAPPY、LZO）
            stmt.execute("SET hive.exec.orc.default.compress=ZLIB");// SET hive.exec.orc.default.compress=SNAPPY;

            // ③ Avro 使用 LZO（也支持 snappy、deflate、bzip2）
            stmt.execute("SET avro.output.codec=lzo"); // SET avro.output.codec=snappy;

            // 创建外部表，指定 HDFS 目录（以 ORC 为例） :插入数据：INSERT OVERWRITE TABLE orc_table SELECT * FROM source_table;
            stmt.execute("DROP TABLE IF EXISTS orc_output_table");
            // ORC 示例（压缩 + 自定义路径）
            stmt.execute(
                    "CREATE EXTERNAL TABLE orc_output_table (id INT, name STRING, age INT) " +
                            "STORED AS ORC " +
                            "LOCATION '/user/hive/warehouse/custom_output/orc_table'"
            );
            // Parquet 示例（压缩 + 自定义 HDFS 路径）:INSERT OVERWRITE TABLE parquet_table SELECT * FROM source_table;
            /*CREATE EXTERNAL TABLE parquet_table (
                    id INT,
                    name STRING,
                    age INT
            )
            STORED AS PARQUET
            LOCATION '/user/hive/warehouse/custom/parquet_output';*/

            // Avro 示例（压缩 + 自定义路径）
/*
CREATE EXTERNAL TABLE avro_table
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.avro.AvroSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.avro.AvroContainerOutputFormat'
LOCATION '/user/hive/warehouse/custom/avro_output'
TBLPROPERTIES (
  'avro.schema.literal' = '{
     "type": "record",
     "name": "User",
     "fields": [
       {"name": "id", "type": "int"},
       {"name": "name", "type": "string"},
       {"name": "age", "type": "int"}
     ]
   }'
);
 */



            // 写入数据（从已存在的源表）
            stmt.execute("INSERT OVERWRITE TABLE orc_output_table SELECT * FROM source_table");

            System.out.println("数据写入成功，压缩格式为 ZLIB，已保存到 HDFS");

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @GetMapping("/create/table")
    public String createTable() {
        Connection connection = HiveDataSourceUtil.createConnection(configPath);
        try {
            // HiveDataSourceUtil.executeSql(connection, SQLS[0]);
            // 查询
            //String queryResult = HiveDataSourceUtil.executeQuery(connection, SQLS[2]);
            String sql = "CREATE EXTERNAL TABLE IF NOT EXISTS hive_std_16LZDR (\n" +
                    "  id STRING,\n" +
                    "  CQ00027_DE00001 STRING,\n" +
                    "  CQ00027_DE00002 STRING \n" +
                    //"  age INT\n" +
                    ")\n" +
                    "STORED BY 'org.apache.hadoop.hive.hbase.HBaseStorageHandler'\n" +
                    "WITH SERDEPROPERTIES (\n" +
                    "  \"hbase.columns.mapping\" = \":key,cf:CQ00027_DE00001,cf:CQ00027_DE00002\"\n" +
                    ")\n" +
                    "TBLPROPERTIES (\n" +
                    "  \"hbase.table.name\" = \"std:std_16LZDR\"\n" +
                    ")";
            HiveDataSourceUtil.executeSql(connection, sql);
            // 自动关闭连接写法
            /*try (Connection conn = HiveDataSourceUtil.createConnection(configPath);
                 Statement stmt = conn.createStatement()) {
                stmt.execute(sql);
            }*/
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }finally {
            HiveDataSourceUtil.closeConnection(connection);
        }

        return "创建hive成功success";
    }
}
